package org.cainiao.servlet.storage.redis.service.impl;

import lombok.RequiredArgsConstructor;
import org.cainiao.servlet.storage.redis.dto.CnEntry;
import org.cainiao.servlet.storage.redis.dto.UserDTO;
import org.redisson.api.*;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import java.time.Duration;
import java.time.LocalDate;
import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * <br />
 * <p>
 * Author: Cai Niao(wdhlzd@163.com)<br />
 */
@Service
@RequiredArgsConstructor
public class RedisServiceImpl {

    private final RedissonClient redissonClient;
    private static final String CACHE_PREFIX = "test:cache:database1:table1:";

    /**
     * 布隆过滤器是一个不可变的数据结构，无法删除其中的记录<br />
     * 因为一旦一个元素被添加到了布隆过滤器中，它就会影响到了位数组的多个位置<br />
     * 其它可能依赖这些位置的元素也可能被影响<br />
     * <p>
     * 如果删除操作不频繁，可以选择定期重新建立布隆过滤器，以清除不再需要的元素<br />
     * 这样做可以保持布隆过滤器的高效性，并防止误判率过高<br />
     * 用空值缓存来做二层防护，防止缓存穿透，控制缓存的过期时间可以与布隆过滤器重建的周期一致，例如每天凌晨进行重建
     *
     * @param userId 用户 ID
     * @return 用户名
     */
    public String getUserName(String userId) {
        // 先查询布隆过滤器，如果没有则不需要执行后边的操作
        RBloomFilter<String> bloomFilter = redissonClient.getBloomFilter(CACHE_PREFIX);
        if (!bloomFilter.contains(userId)) {
            return null;
        }

        // 查询缓存
        String cacheKey = CACHE_PREFIX + userId;
        RBucket<String> bucket = redissonClient.getBucket(cacheKey);
        String cachedResult = bucket.get();
        if (StringUtils.hasText(cachedResult)) {
            return cachedResult;
        }

        String databaseResult = loadFromDatabase(userId);
        if (!StringUtils.hasText(databaseResult)) {
            // 布隆过滤器命中，但数据库中没有，说明布隆过滤器发生了哈希冲突，此时通过空值缓存来做第二道防线，防止缓存穿透
            bucket.set("", Duration.ofSeconds(60));
            return null;
        }

        // 数据库中存在值，设置缓存并返回
        bucket.set(databaseResult, Duration.ofSeconds(60));
        return databaseResult;
    }

    public void setUserName(String userId, String userName) {
        RBloomFilter<String> bloomFilter = redissonClient.getBloomFilter(CACHE_PREFIX);
        boolean isAdded = bloomFilter.add(userId);
        if (!isAdded) {
            // 如果元素已经在布隆过滤器中，则 add 方法返回 false
            System.out.println("The ID might already be in the bloom filter or an error occurred.");
        }

        // 入库
        setUserNameToDatabase(userId, userName);
    }

    private final ConcurrentHashMap<String, Lock> localLocks = new ConcurrentHashMap<>();

    /**
     * 重建布隆过滤器<br />
     * 需要保证在此期间，没有数据新增入数据库，可以通过在本方法上加上串行事务来实现
     *
     * @param filterName 过滤器名称
     * @param data       数据
     */
    @Transactional(isolation = Isolation.SERIALIZABLE)
    public void rebuildBloomFilter(String filterName, Collection<String> data) {
        if (!StringUtils.hasText(filterName)) {
            return;
        }
        if (hasRebuiltToday(filterName)) {
            // 今天已经重建过了，不需要再次重建
            return;
        }
        // 本地锁，保证每个 JVM 进程只有一个线程向 Redis 发起 I/O
        Lock localLock = localLocks.computeIfAbsent(filterName, key -> new ReentrantLock());
        if (!localLock.tryLock()) {
            // 未获取到锁说明已经有人在重建了，自己快速失败即可
            return;
        }
        try {
            // 分布式锁保证分布式环境（如多个副本）只有一个线程进行重建
            Lock distributedLock = redissonClient.getLock(filterName + "_lock");
            if (!distributedLock.tryLock()) {
                // 未获取到锁说明已经有人在重建了，自己快速失败即可
                return;
            }
            try {
                if (hasRebuiltToday(filterName)) {
                    /*
                     * 重新校验下状态，因为在第一次判断状态时，可能有并发的线程都判断为还没有重建
                     * 而其中一些线程可能因为直到进入临界区的线程释放资源都没有被操作系统调度
                     * 因此并没有因为 tryLock 而快速失败，最终这些线程会进入到这里
                     * 但实际上今天已经重建过了，所以需要在这里再次验证状态，避免重复重建布隆过滤器
                     */
                    return;
                }

                // 构建新的布隆过滤器
                RBloomFilter<String> newBloomFilter = redissonClient.getBloomFilter("temp-" + filterName);
                newBloomFilter.tryInit(1000000L, 0.01);
                for (String element : data) {
                    newBloomFilter.add(element);
                }

                // 替换现有的布隆过滤器
                RBloomFilter<String> oldBloomFilter = redissonClient.getBloomFilter(filterName);
                String oldFilterName = oldBloomFilter.getName();
                redissonClient.getKeys().delete(oldFilterName);
                newBloomFilter.rename(oldFilterName);

                // 设置标志，表示今天已经重建过了，避免重复重建布隆过滤器
                redissonClient.getBucket(getMarkKey(filterName)).set(LocalDate.now().toString());
            } finally {
                distributedLock.unlock();
            }
        } finally {
            localLock.unlock();
        }
    }

    private boolean hasRebuiltToday(String filterName) {
        LocalDate lastRebuiltDate = getLastRebuiltDate(filterName);
        LocalDate today = LocalDate.now();
        return lastRebuiltDate != null && lastRebuiltDate.equals(today);
    }

    private LocalDate getLastRebuiltDate(String filterName) {
        String lastRebuiltDate = (String) redissonClient.getBucket(getMarkKey(filterName)).get();
        return lastRebuiltDate != null ? LocalDate.parse(lastRebuiltDate) : null;
    }

    private String getMarkKey(String filterName) {
        return filterName + "_rebuilt_date";
    }

    private String loadFromDatabase(String userId) {
        return userId + " name";
    }

    private void setUserNameToDatabase(String userId, String userName) {
        System.out.println(userId);
        System.out.println(userName);
    }

    /**
     * Redis based distributed unbounded BlockingQueue object for Java implements java.util.concurrent.BlockingQueue interface.<br />
     * This object is fully thread-safe.
     * <p>
     * It has Async, Reactive and RxJava3 interfaces.
     */
    public void blockingQueueTest() throws InterruptedException {
        RBlockingQueue<UserDTO> queue = redissonClient.getBlockingQueue("anyQueue");

        System.out.println(queue.offer(new UserDTO()));

        UserDTO obj = queue.peek();
        UserDTO obj2 = queue.poll();
        UserDTO obj3 = queue.poll(10, TimeUnit.MINUTES);
    }

    /**
     * Redis based distributed BoundedBlockingQueue for Java implements java.util.concurrent.BlockingQueue interface.<br />
     * BoundedBlockingQueue size limited by Redis to 4 294 967 295 elements.<br />
     * This object is fully thread-safe.
     * <p>
     * Queue capacity should be defined once by trySetCapacity() method before the usage:
     */
    public void boundedBlockingQueueTest() throws InterruptedException {
        RBoundedBlockingQueue<UserDTO> queue = redissonClient.getBoundedBlockingQueue("anyQueue");
        // returns `true` if capacity set successfully and `false` if it already set.
        queue.trySetCapacity(2);

        System.out.println(queue.offer(new UserDTO(1)));
        System.out.println(queue.offer(new UserDTO(2)));
        // will be blocked until free space available in queue
        queue.put(new UserDTO());

        UserDTO obj = queue.peek();
        UserDTO someObj = queue.poll();
        UserDTO ob = queue.poll(10, TimeUnit.MINUTES);

        /*
         * poll, pollFromAny, pollLastAndOfferFirstTo and take methods will be resubscribed automatically
         * during reconnection to Redis server or Redis server failover.
         */
    }

    /**
     * Java implementation of Redis based BlockingDeque implements java.util.concurrent.BlockingDeque interface.<br />
     * This object is fully thread-safe.
     * <p>
     * It has Async, Reactive and RxJava3 interfaces.
     */
    public void blockingDequeTest() throws InterruptedException {
        RBlockingDeque<Integer> deque = redissonClient.getBlockingDeque("anyDeque");
        deque.putFirst(1);
        deque.putLast(2);
        Integer firstValue = deque.takeFirst();
        Integer lastValue = deque.takeLast();
        Integer firstValue2 = deque.pollFirst(10, TimeUnit.MINUTES);
        Integer lastValue2 = deque.pollLast(3, TimeUnit.MINUTES);
    }

    /**
     * Redis based distributed BlockingFairQueue for Java implements java.util.concurrent.BlockingQueue interface.<br />
     * This object is fully thread-safe.
     * <p>
     * When queue consumers in different parts of network: some of them closer to redis and some further.<br />
     * "further" consumers will get lower amount of messages from queue due to network delays.<br />
     * In turn "closer" consumers will get higher amount and this could lead to client overloading.
     * <p>
     * Blocking queue with fair polling guarantees access order for poll and take methods
     * and allows to get uniformly distributed consumption.
     */
    public void blockingFairQueueTest() {
//        RBlockingFairQueue queue = redissonClient.getBlockingFairQueue("myQueue");
//        queue.offer(new SomeObject());
//
//        SomeObject element1 = queue.peek();
//        SomeObject element2 = queue.poll();
//        SomeObject element3 = queue.poll(10, TimeUnit.MINUTES);
//        SomeObject element4 = queue.take();
        // This feature available only in Redisson PRO edition.(https://redisson.pro/)
    }

    /**
     * Redis based DelayedQueue object for Java allows to transfer each element to destination queue with specified delay.<br />
     * Destination queue could be any queue implemented RQueue interface.<br />
     * This object is fully thread-safe.
     * <p>
     * Could be useful for exponential backoff strategy used for message delivery to consumer.<br />
     * If application is restarted, an instance of delayed queue should created in order for the pending items
     * to be added to the destination queue.
     */
    public void delayedDequeTest() throws InterruptedException {
        RBlockingQueue<String> destinationQueue = redissonClient.getBlockingQueue("anyQueue");
        RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(destinationQueue);
        // move object to destination Queue in 10 seconds
        delayedQueue.offer("msg1", 10, TimeUnit.SECONDS);
        // move object to destination Queue in 1 minutes
        delayedQueue.offer("msg2", 1, TimeUnit.MINUTES);


        // msg1 will appear in 10 seconds
        String msg1 = destinationQueue.poll(15, TimeUnit.SECONDS);
        System.out.printf("msg1 = %s%n", msg1);

        /*
         * 2 秒后会打印 msg2 为 null
         * 延迟队列只是负责在延迟特定时间后向目标队列放入数据
         * 但并不会影响目标队列行为，例如阻塞队列的 poll() 方法行为
         * 如果队列中没有数据，则等 2 秒，如果还是没有则返回 null
         * 如果在等待时间期间有数据了，则会返回得到的数据
         */
        String msg2 = destinationQueue.poll(2, TimeUnit.SECONDS);
        System.out.printf("msg2 = %s%n", msg2);

        /*
         * Object should be destroyed if it not used anymore,
         * but it's not necessary to call destroy method if Redisson goes shutdown.
         */
        delayedQueue.destroy();
    }

    /**
     * Java implementation of Redis based PriorityQueue implements java.util.Queue interface.<br />
     * Elements are ordered according to natural order of java.lang.Comparable interface or defined java.util.Comparator.<br />
     * This object is fully thread-safe.
     * <p>
     * Use trySetComparator() method to define own java.util.Comparator.
     * <p>
     * Code example:
     */
    public void priorityQueueTest() {
        RPriorityQueue<CnEntry> queue = redissonClient.getPriorityQueue("anyQueue");
        queue.add(new CnEntry("b", 1));
        queue.add(new CnEntry("c", 1));
        queue.add(new CnEntry("a", 1));

        // Entry [a:1]
        CnEntry e = queue.poll();
        // Entry [b:1]
        CnEntry e2 = queue.poll();
        // Entry [c:1]
        CnEntry e3 = queue.poll();
    }
}
